package com.wyd.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class KafkaConfig {

  @Value("${kafka.topic.order}")
  private String orderTopic;

  /**
   * 创建topic
   */
  @Bean
  public boolean createOrderTopic(){
    log.info("kafka创建topic：{}",orderTopic);
    NewTopic topic = new NewTopic(orderTopic,2,(short)1);
    return true;
  }

  /**
   * MANUAL_IMMEDIATE 手动调用Acknowledgment.acknowledge()后立即提交
   * @param consumerFactory
   * @return
   */
  @Bean("manualImmediateListenerContainerFactory")
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualImmediateListenerContainerFactory(
          ConsumerFactory<String, String> consumerFactory) {

    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.getContainerProperties().setPollTimeout(1500);
    factory.setBatchListener(true);
    //配置手动提交offset
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    return factory;
  }


}